package com.simple.trade.service.vm.impl;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Joiner;
import com.simple.trade.holder.PayContext;
import com.simple.trade.service.vm.NotifyService;
import com.simple.common.entity.ReqContext;
import com.simple.common.entity.po.Notify;
import com.simple.common.entity.po.NotifyChannel;
import com.simple.common.entity.po.sys.Application;
import com.simple.common.entity.po.sys.Trade;
import com.simple.common.entity.req.vm.pay.NotifyReq;
import com.simple.ds.service.NotifyService2;
import com.simple.ds.service.TradeApi;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.time.Clock;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author yasud
 * @date 2020年10月12日 11:46
 * @description
 */
@Slf4j
@Service
public class NotifyServiceImpl implements NotifyService {

    private static final String MERCHANT_NOTIFY_EXCHANGE = "merchant_notify_exchange";

    private static final String MERCHANT_NOTIFY_QUEUE = "merchant_notify_queue";

    private static final int MAX_NOTIFY_COUNT = 6;

    private static final String FINAL_SUCCESS_WORD = "0";

    @Autowired
    ObjectMapper objectMapper;

    @Autowired
    NotifyService2 notifyApi;

    @Autowired
    Clock clock;

    @Autowired
    TradeApi tradeApi;

    private ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(10);

    /**
     * 创建消息
     *
     * @param notifyChannel
     * @param trade
     * @return
     * @throws JsonProcessingException
     */
    @Override
    public Notify createNotify(NotifyChannel notifyChannel, Trade trade) throws JsonProcessingException {
        NotifyReq notifyReq = generateNotifyReq(trade);
        ReqContext context = PayContext.current().getContext();
        Notify notify = new Notify();
        notify.setAppId(trade.getAppId());
        notify.setMerchantName(context.getMerchant().getName());
        notify.setMerchantNo(context.getMerchant().getMerchantNo());
        notify.setExpName(PayContext.current().exp().getName());
        notify.setCount(0);
        notify.setNotifyUrl(trade.getAsyncUrl());
        notify.setStatus(Notify.Status.NEW);
        notify.setLastNotifyTime(new Date());
        notify.setTradeNo(trade.getId());
        notify.setMerchantOrderNo(trade.getMerchantOrderNo());
        notify.setMessage(objectMapper.writeValueAsString(notifyReq));

        notify = notifyApi.save_(notify);
        sendToAmqpNow(notify);
        return notify;
    }

    /**
     * 发送
     *
     * @param notify
     */
    @SneakyThrows
    private void sendToAmqpDelay(Notify notify) {
        scheduledExecutorService.schedule(() -> {
            reSend(notify);
        }, 1, TimeUnit.MINUTES);
    }

    private void reSend(Notify notify) {
        try {
            recv(notify);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 立刻放到mq
     *
     * @param notify
     */
    @SneakyThrows
    private void sendToAmqpNow(Notify notify) {
        scheduledExecutorService.execute(() -> {
            reSend(notify);
        });
    }

    /**
     * 接收
     *
     * @param notify
     */
    @SneakyThrows
    public void recv(Notify notify) {
        log.info("收到消息{}", notify.getId());
        Trade trade = tradeApi.findById_(notify.getTradeNo());
        PayContext.current().iam(trade);
        sendToMerchant(notify);
    }


    @Override
    @SneakyThrows
    public void sendToMerchant(Notify notify) {
        String notifyUrl = notify.getNotifyUrl();
        if (StringUtils.isBlank(notifyUrl)) {
            log.info("渠道回调地址为空");
            return;
        }

        String body = "";
        try {
            HttpRequest post = HttpUtil.createPost(notifyUrl);
            post.contentType(MediaType.APPLICATION_JSON_UTF8_VALUE);
            post.body(notify.getMessage());

            HttpResponse rsp = post.execute();
            body = rsp.body();
            log.info("消息{}通知{}返回{}", notify.getId(), notify.getNotifyUrl(), body);
        } catch (Exception ex) {
            log.info("消息{}通知异常", ex.getMessage());
            ex.printStackTrace();
        }

        notify.setResult(StringUtils.abbreviate(body, 100));
        notify.setLastNotifyTime(new Date());
        notify.setStatus(Notify.Status.PROCESSED);
        String finalBody = body;
        ObjectNode jsonNodes = Mono.fromCallable(() -> objectMapper.readValue(finalBody, ObjectNode.class))
                .onErrorReturn(new ObjectNode(JsonNodeFactory.instance))
                .block();
        String code = "1";
        JsonNode jsonNode = jsonNodes.get("code");
        if (Objects.nonNull(jsonNode)) {
            code = jsonNode.asText();
        }

        if (FINAL_SUCCESS_WORD.equalsIgnoreCase(code)) {
            notify.setStatus(Notify.Status.SUCCESS);
            log.info("消息{}通知成功，现在结束", notify.getId());
        } else {
            if (notify.getCount() + 1 < MAX_NOTIFY_COUNT) {
                log.info("消息{}通知失败, 现在次数{}", notify.getId(), notify.getCount());
                notify.setCount(notify.getCount() + 1);
                sendToAmqpDelay(notify);
            } else {
                notify.setStatus(Notify.Status.FAILED);
                notify.setCount(notify.getCount() + 1);
                log.info("消息{}已达最大通知数量,通知失败", notify.getId());
            }
        }
        notifyApi.updateById_(notify);
    }

    /**
     * 生成签名
     *
     * @param req
     * @return
     * @throws Exception
     */
    private String sign(NotifyReq req) throws Exception {
        Application app = PayContext.current().app();
        Map<String, Object> describe = PropertyUtils.describe(req);
        describe.remove("remark");
        describe.remove("payMsg");
        List<Map.Entry<String, Object>> collect = describe.entrySet()
                .stream()
                .filter(stringObjectEntry -> Objects.nonNull(stringObjectEntry.getValue()))
                .sorted(Comparator.comparing(stringObjectEntry -> stringObjectEntry.getKey()))
                .collect(Collectors.toList());
        String str = Joiner.on("&").withKeyValueSeparator("=").join(collect);
        String signStr = str + app.getSecretKey();
        log.info("订单[{}]通知商户结果{},待签名串为{}", req.getSerialNo(), "签名中", signStr);
        String hex = DigestUtils.sha256Hex(signStr);
        log.info("订单[{}]通知商户结果{},签名结果为{}", req.getSerialNo(), "签名中", hex);
        return hex;
    }

    @SneakyThrows
    public static void main(String[] args) {
        String str = "{\"appId\":\"EvenyLXUhUB4\",\"orderNo\":\"190497991\",\"serialNo\":\"1379937278906040322\",\"amount\":\"414.19\",\"currency\":\"CNY\",\"remark\":null,\"payMsg\":\"payment success\",\"payStatus\":\"SUCCESS\",\"ts\":\"1617758780564\"}";
        Map<String, Object> describe = new ObjectMapper().readValue(str, Map.class);
        describe.remove("remark");
        describe.remove("payMsg");
        List<Map.Entry<String, Object>> collect = describe.entrySet()
                .stream()
                .filter(stringObjectEntry -> Objects.nonNull(stringObjectEntry.getValue()))
                .sorted(Comparator.comparing(stringObjectEntry -> stringObjectEntry.getKey()))
                .collect(Collectors.toList());
        String strss = Joiner.on("&").withKeyValueSeparator("=").join(collect);
        String signStr = strss + "meqpUKqvJU";
        String hex = DigestUtils.sha256Hex(signStr);
        System.out.println(hex);
    }

    /**
     * 查询这个交易id是否有回调
     *
     * @param tradeId
     * @return
     */
    @Override
    public int countNotify(Long tradeId) {
        int count = notifyApi.countByTradeNo(tradeId);
        return count;
    }

    @Override
    public Notify getNotifyByTradeNo(Long tradeNo) {
        return notifyApi.getNotifyByTradeNo(tradeNo);
    }

    /**
     * 这个是给商户的
     *
     * @param trade
     * @return
     */
    @SneakyThrows
    private NotifyReq generateNotifyReq(Trade trade) {
        NotifyReq notifyReq = new NotifyReq();
        notifyReq.setAppId(trade.getAppId());
        notifyReq.setAmount(trade.getAmount());
        notifyReq.setCurrency(trade.getCurrency());
        notifyReq.setPayStatus(trade.getStatus());
        notifyReq.setSerialNo(trade.getId());
        notifyReq.setPayMsg(trade.getMsg());
        notifyReq.setOrderNo(trade.getMerchantOrderNo());
        notifyReq.setRemark(trade.getOrderDesc());
        notifyReq.setTs(clock.millis());
        notifyReq.setSignInfo(sign(notifyReq));
        return notifyReq;
    }
}
